package com.atguigu.ting.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.ting.realtime.app.BaseApp;
import com.atguigu.ting.realtime.bean.PayItemBean;
import com.atguigu.ting.realtime.common.Constant;
import com.atguigu.ting.realtime.util.AtguiguUtil;
import com.atguigu.ting.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/4/14 12:49
 */
public class DwsPayItemWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsPayItemWindow().init(
            40014,
            2,
            "DwsPayItemWindow",
            Constant.TOPIC_DWD_TRADE_PAY_SUCCESS
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
       
        stream
            .map(new MapFunction<String, PayItemBean>() {
                @Override
                public PayItemBean map(String value) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    return PayItemBean.builder()
                        .orderDetailId(obj.getString("order_detail_id"))
                        .payItemName(obj.getString("item_type_name"))
                        .payAmount(obj.getBigDecimal("split_order_amount"))
                        .ts(obj.getLong("payment_time"))
                        .build();
                }
            })
            // 由于支付表数据来源于下单详情表,下单详情表有重复数据,所以这里需要按照详情 id 去重.
            .keyBy(PayItemBean::getOrderDetailId)
            .process(new KeyedProcessFunction<String, PayItemBean, PayItemBean>() {
                private ValueState<PayItemBean> state;
    
                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<PayItemBean> desc =
                        new ValueStateDescriptor<>("beanState", PayItemBean.class);
                    StateTtlConfig ttlConfig = new StateTtlConfig.Builder(org.apache.flink.api.common.time.Time.seconds(10))
                        .updateTtlOnCreateAndWrite()
                        .build();
        
                    desc.enableTimeToLive(ttlConfig);
                    state = getRuntimeContext().getState(desc);
                }
    
                @Override
                public void processElement(PayItemBean currentBean,
                                           Context ctx,
                                           Collector<PayItemBean> out) throws Exception {
                    PayItemBean lastBean = state.value();
                    if (lastBean == null) { // 当前详情 id 第一条数据
                        out.collect(currentBean); // 直接输出
                    } else { // 不是第一条
                        // 用新的数据, 减去状态中的数据, 输出
                        lastBean.setPayAmount(currentBean.getPayAmount().subtract(lastBean.getPayAmount()));
            
                        out.collect(lastBean);
                    }
                    state.update(currentBean);  // 更新状态
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<PayItemBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))
            )
            .keyBy(PayItemBean::getPayItemName)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<PayItemBean>() {
                    @Override
                    public PayItemBean reduce(PayItemBean value1,
                                                  PayItemBean value2) throws Exception {
                        value1.setPayAmount(value1.getPayAmount().add(value2.getPayAmount()));
                        return value1;
                    }
                },
                new ProcessWindowFunction<PayItemBean, PayItemBean, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<PayItemBean> elements,
                                        Collector<PayItemBean> out) throws Exception {
                        PayItemBean bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(ctx.window().getEnd()));
                    
                        bean.setTs(System.currentTimeMillis());
                    
                        out.collect(bean);
                    }
                }
            )
            .addSink(FlinkSinkUtil.getClickHouseSink("dws_pay_item_window", PayItemBean.class));
        
    }
    
}
